/*
 * 文件名： RespondDataTask.java
 *
 * 创建日期： 2016年12月5日
 *
 * Copyright(C) 2016, by <a href="mailto:liws@xingyuanauto.com">liws</a>.
 *
 * 原始作者: liws
 *
 */
package com.moli.iov.gateway.server.task;

import com.moli.iov.gateway.server.NettyChannelManager;
import com.moli.iov.gateway.server.VinThreadPoolExecutor;
import com.moli.iov.gateway.server.model.KafkaSendData;
import com.moli.iov.gateway.util.logger.GatewayLogger;
import com.moli.iov.protocol.codec.DataWrapper;
import com.moli.iov.protocol.codec.DownStreamData;
import com.moli.iov.protocol.enums.CommandTag;
import com.moli.iov.protocol.enums.DateTimeTag;
import com.moli.iov.protocol.enums.ResponseTag;
import com.moli.iov.protocol.util.BccUtils;
import com.moli.iov.util.ByteArrayUtil;
import com.moli.iov.util.PropertiesUtils;
import com.moli.iov.util.SpringContextUtils;
import com.moli.iov.util.StringUtil;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import java.util.Date;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 应答消息 多线程处理Task
 *
 * @author <a href="mailto:liws@xingyuanauto.com">liws</a>
 * @version $Revision$
 * @since 2016年12月5日
 */
public class DownStreamDataTask implements Runnable {

    private static final Logger logger = LoggerFactory.getLogger(DownStreamDataTask.class);

    private DownStreamData dsd;

    private String vinLogPath;

    private String kafkaDownGetwayReponseTopic;

    /**
     * 命令索引
     */
    private static final int COMMAND_INDEX = 2;
    /**
     * kafka数据发送线程池
     */
    private VinThreadPoolExecutor kafkaSendThreadPool;

    public DownStreamDataTask(DownStreamData dsd) {
        this.dsd = dsd;
        vinLogPath = PropertiesUtils.getProperty("vin.log.path");
        kafkaDownGetwayReponseTopic = PropertiesUtils.getProperty("kafka.down.getway.reponse.topic");
        kafkaSendThreadPool = SpringContextUtils.getBean("kafkaSendThreadPool");
    }

    /**
     * @see java.lang.Runnable#run()
     */
    @Override
    public void run() {
        try {
            if (dsd == null || dsd.vin == null) {
                logger.error("下行,dsd参数为空或vin为空");
                return;
            }
            byte[] dataBytes = dsd.dataBytes;
            if (dataBytes == null || dataBytes.length < COMMAND_INDEX) {
                logger.error("下行,dsd参数中无有效报文");
                return;
            }
            String vin = dsd.vin;
            /**************方便压测临时代码***************/
//            byte[] eol = {0x24};
//
//            byte[] ret = new byte[eol.length+dataBytes.length];
//            System.arraycopy(dataBytes,0,ret,0,dataBytes.length);
//            System.arraycopy(eol,0,ret,dataBytes.length,eol.length);
//            dsd.dataBytes = ret;
            /***************方便压测临时代码*************/
            final Channel channel;
            if (ResponseTag.REGISTER_RES.getV() == dataBytes[COMMAND_INDEX]) {//注册报文
                channel = NettyChannelManager.getInstance().getRegisterChannelByVin(vin);
            } else {
                channel = NettyChannelManager.getInstance().getLoginChannelByVin(vin);
            }
            if (channel == null) {
                logger.info("下行，未找到有效的channel,vin:{}", vin);
                return;
            }
//            if(!channel.isActive()){
//                logger.info("下行，channel已经关闭,vin:{}", vin);
//                return;
//            }
            if (channel.isWritable()) {
                channel.writeAndFlush(dataBytes).addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture future) throws Exception {
                        if (!future.isSuccess()) {
                            logger.warn("下发数据失败， msg:{}", future.cause());
                        } else {
                            if (CommandTag.VEHICLE_LOOUT.getV() == dataBytes[COMMAND_INDEX]) {//登出发送报文后关闭通道
//                                NettyChannelManager.getInstance().removeChannel(channel);
                                channel.close();
                            }
                        }
                    }
                });
            } else {
                try {
                    channel.writeAndFlush(dataBytes).sync();
                    logger.info("publish macdonaldMsg,sended. remoteAddress:" + channel.remoteAddress() + "");
                } catch (InterruptedException e) {
                    logger.error("write and flush msg exception. ", e);
                }
            }
            GatewayLogger.getLogger(vin, vinLogPath).info("[SEND]:" + ByteArrayUtil.toHexString(dataBytes));
            logger.info("[下行报文][vin=" + vin + "][" + ByteArrayUtil.toHexString(dataBytes) + "]");
            //发成功，网关响应上层
            if (StringUtil.isNotBlank(dsd.serialNo)) {
                gatewayResponse(vin, dsd.serialNo, dataBytes);
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
    }

    /**
     * 下发成功，网关响应上层
     *
     * @param serialNo
     */
    public void gatewayResponse(String vin, String serialNo, byte[] bytes) {
        logger.info("下发成功，网关响应上层 入口 vin:{}  流水号: {}, 数据：{}", vin, serialNo, bytes);
        DataWrapper dataWrapper = new DataWrapper(serialNo, bytes, new Date());
        dataWrapper.setCommand(bytes[COMMAND_INDEX]);
        KafkaSendData kafkaSendData = new KafkaSendData(kafkaDownGetwayReponseTopic, vin, dataWrapper);
        kafkaSendThreadPool.execute(vin, new KafkaSendTask(kafkaSendData));
    }

    public static byte[] resetVIN1() {
        String str = "2323D9FE4C4E425343423346334A5730303030303101001D";
        byte[] logout = ByteArrayUtil.hexStringToBytes(str);
        String str1 = "006a012001114c4e425343423346334a57303030303032";
        byte[] s1 = ByteArrayUtil.hexStringToBytes(str1);
        byte[] toBytes = new DateTimeTag().toBytes();
        byte[] ret = new byte[logout.length + toBytes.length + s1.length];
        System.arraycopy(logout, 0, ret, 0, logout.length);
        System.arraycopy(toBytes, 0, ret, logout.length, toBytes.length);
        System.arraycopy(s1, 0, ret, logout.length + toBytes.length, s1.length);
        byte[] m = BccUtils.bccBodyEncode(ret);
        return m;
    }
}
